package buaa.qel.rpc.server.core;

import buaa.qel.rpc.common.annotation.RpcService;
import buaa.qel.rpc.server.registry.ServiceRegistry;
import buaa.qel.rpc.common.util.ServiceKeyMaker;
import buaa.qel.rpc.common.util.ThreadPoolUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.reflections.Reflections;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * RpcServer类
 * rpc服务器启动流程：
 * 1）构造方法参数：服务器地址，注册地址（即zookeeper服务器地址），服务接口所在的包（全限定名）
 * 2）构造方法：扫描接口包，将服务接口信息全部通过反射获取class类，并加入serviceMap中
 * 3）服务器启动：初始化服务器线程，初始化服务器的线程池（用来执行服务端的方法，获取结果）
 * 4）启动netty服务器对象，将线程池和服务表传递给channelInitializer的构造函数，用于初始化requestHandler对象
 * 5）向zookeeper注册服务信息
 */
@Slf4j
public class RpcServer extends Server{
    private Thread serverThread;
    private String serverAddress;
    private ServiceRegistry serviceRegistry;
    private Map<String, Object> serviceMap = new HashMap<>();

    public RpcServer(String serverAddress, String registryAddress,String packageScan) {
        this.serverAddress = serverAddress;
        this.serviceRegistry = new ServiceRegistry(registryAddress);

        Reflections reflections = new Reflections(packageScan);
        Set<Class<?>> set = reflections.getTypesAnnotatedWith(RpcService.class);
        try {
            for (Class<?> serviceBean : set) {
                Object newInstance = Class.forName(serviceBean.getName()).newInstance();
                String interfaceName = serviceBean.getInterfaces()[0].getName();
                String serviceName = serviceBean.getAnnotation(RpcService.class).serviceName();
                addService(interfaceName, serviceName, newInstance);
            }
        } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
            throw new ExceptionInInitializerError(e);
        }
    }

    protected void addService(String interfaceName, String serviceName, Object serviceBean) {
        log.info("Adding service, interfaceName: {}, serviceName: {}, bean：{}", interfaceName, serviceName, serviceBean);
        String serviceKey = ServiceKeyMaker.makeServiceKey(interfaceName, serviceName);
        serviceMap.put(serviceKey, serviceBean);
    }

    public void start() {
        serverThread = new Thread(new Runnable() {
            ThreadPoolExecutor threadPoolExecutor = ThreadPoolUtil.makeServerThreadPool(RpcServer.class.getSimpleName());

            @Override
            public void run() {
                EventLoopGroup bossGroup = new NioEventLoopGroup();
                EventLoopGroup workerGroup = new NioEventLoopGroup();
                try {
                    ServerBootstrap bootstrap = new ServerBootstrap();
                    bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                            .childHandler(new RpcServerChannelInitializer(serviceMap, threadPoolExecutor))
                            .option(ChannelOption.SO_BACKLOG, 128)
                            .childOption(ChannelOption.SO_KEEPALIVE, true);

                    String[] array = serverAddress.split(":");
                    String host = array[0];
                    int port = Integer.parseInt(array[1]);
                    ChannelFuture future = bootstrap.bind(host, port).sync();

                    if (serviceRegistry != null) {
                        serviceRegistry.registerService(host, port, serviceMap);
                    }
                    log.info("Server started on port {}", port);
                    /**
                     * 该代码的作用为：让服务器线程在绑定端口后到监听到关闭事件之前，一直保持等待状态
                     * 当监听到关闭事件，则在sync()执行后进入finally代码块
                     */
                    future.channel().closeFuture().sync();
                } catch (Exception e) {
                    if (e instanceof InterruptedException) {
                        log.info("Rpc server remoting server stop");
                    } else {
                        log.error("Rpc server remoting server error", e);
                    }
                } finally {
                    try {
                        if(serviceRegistry != null){
                            serviceRegistry.unregisterService();
                        }
                        workerGroup.shutdownGracefully();
                        bossGroup.shutdownGracefully();
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                }
            }
        });
        serverThread.start();
    }

    public void stop() {
        if (serverThread != null && serverThread.isAlive()) {
            serverThread.interrupt();
        }
    }
}
